package com.bigdata.spark.mallapp_realtime

import java.lang

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.bigdata.spark.util.{MyKafkaUtil, OffsetManagerUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable.ListBuffer

object mall_log_PV_CountByShop {
  def main(args: Array[String]): Unit = {
    val actions_count: SparkConf = new SparkConf().setMaster("local[*]").setAppName("actions_count")
    val ssc: StreamingContext = new StreamingContext(actions_count,Seconds(5))


    var topic:String = "gmall_event_0523"
    var groupId = "action_count"

    val offsetMap: Map[TopicPartition, Long] = OffsetManagerUtil.getOffset(topic, groupId)

    var recordDStream: InputDStream[ConsumerRecord[String, String]] = null

    if(offsetMap.size>0&& offsetMap!=null){
      recordDStream = MyKafkaUtil.getKafkaStream(topic,ssc,offsetMap,groupId)
    }else{
      recordDStream = MyKafkaUtil.getKafkaStream(topic,ssc,groupId)
    }


    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]

    val KafkaDS: DStream[ConsumerRecord[String, String]] = recordDStream.transform {
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }

    }


    val actarr: DStream[(String, String)] = KafkaDS.flatMap(
      kafkadata => {

        val str: String = kafkadata.value()
        val jsonlog: JSONObject = JSON.parseObject(str)
        val common = jsonlog.getJSONObject("common")
        val actions = jsonlog.getJSONArray("actions")
        val displays: JSONArray = jsonlog.getJSONArray("displays")
        val page: JSONObject = jsonlog.getJSONObject("page")
        val err: JSONObject = jsonlog.getJSONObject("err")
        val ts: lang.Long = jsonlog.getLong("ts")

        val actarr = new ListBuffer[(String, String)] //存储动作id，目标id
        //       获取元素（使用括号访问(索引值)）
        //添加元素（+=）
        //追加一个列表（++=）
        //更改元素（使用括号获取元素，然后进行赋值）
        //删除元素（-=）
        //转换为List（toList）
        //转换为Array（toArray）
        //        if(actions!=null){           //统计十分钟内所有动作id，并且
        //
        if (actions!=null&& actions.size()>0) { //对本json中的动作数据遍历。


          for (i <- 0 to actions.size() - 1) {
            actarr.append((actions.getJSONObject(i).getString("action_id"), actions.getJSONObject(i).getString("item")))
            //println("动作列表",(actions.getJSONObject(i).getString("action_id"), actions.getJSONObject(i).getString("item")))

          }
        }


        actarr




      }
    )

    val wordToOne: DStream[((String, String), Int)] = actarr.map((_, 1))

    val windowDS: DStream[((String, String),Int)] = wordToOne.window(Seconds(30), Seconds(30))

    val wordToCount = windowDS.reduceByKey(_+_)

    //wordToCount.print()
    wordToCount.foreachRDD{
      rdd=>{

        rdd.foreach(
          record=>{
            println(record)
          }
        )

        println("第一层")
        println(rdd,rdd.getClass)
        //提交偏移量到Redis中，以分区为单位。
        //println("提交偏移量到Redis中，以分区为单位。")
        OffsetManagerUtil.saveOffset(topic,groupId,offsetRanges)

      }

    }


    ssc.start()

    ssc.awaitTermination()




  }
}
